// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: taekwon

#include "coin2/exchange/base/order/private_ws_subscriber.h"

#include <algorithm>
#include <atomic>
#include <cstdlib>
#include <functional>
#include <iostream>
#include <map>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <utility>
#include <vector>

#include <boost/algorithm/string.hpp>
#include <glog/logging.h>
#include <nlohmann/json.hpp>

#include "coin2/exchange/base/api_base/raw_subscriber.h"
#include "coin2/exchange/base/api_util/zlib_decompress.h"
#include "coin2/exchange/base/executor/live_executor.h"
#include "coin2/exchange/base/order/parser.h"
#include "coin2/exchange/base/order/processor.h"
#include "coin2/exchange/bitflyer_futures/api_util/auth_key.h"
#include "coin2/exchange/bitget_base/api_util/auth_key.h"
#include "coin2/exchange/bitmart/api_util/auth_key.h"
#include "coin2/exchange/bitmex/api_util/auth_key.h"
#include "coin2/exchange/bittrex/api_util/auth_key.h"
#include "coin2/exchange/bybit/api_util/auth_key.h"
#include "coin2/exchange/bybit_futures/api_util/auth_key.h"
#include "coin2/exchange/crypto_common/api_util/auth_key.h"
#include "coin2/exchange/dydx_futures/api_util/auth_key.h"
#include "coin2/exchange/ftx_common/api_util/auth_key.h"
#include "coin2/exchange/gateio/api_util/auth_key.h"
#include "coin2/exchange/gdax/api_util/auth_key.h"
#include "coin2/exchange/gopax/api_util/auth_key.h"
#include "coin2/exchange/huobi_futures/api_util/auth_key.h"
#include "coin2/exchange/kucoin/api_util/auth_key.h"
#include "coin2/exchange/maicoin/api_util/auth_key.h"
#include "coin2/exchange/okex_common/api_util/auth_key.h"
#include "coin2/exchange/phemex_base/api_util/auth_key.h"
#include "coin2/exchange/prex_futures/api_util/auth_key.h"

namespace {

inline std::string replace_string(std::string str, const std::string& from, const std::string& to) {
  size_t start_pos = str.find(from);
  if (start_pos == std::string::npos) {
    return str;
  }
  str.replace(start_pos, from.length(), to);
  return str;
}

inline std::string get_pong(const std::string& str) { return replace_string(str, "ping", "pong"); }

}  // namespace

namespace coin2::exchange::base::order {

BasePrivateWsConnection::BasePrivateWsConnection(
    const PrivateWsSubscriberSpec& spec,
    std::unique_ptr<BasePrivateWsSubscriberPolicy> policy,
    std::function<void()> fail_callback,
    boost::asio::io_context* ioc,
    const MarketExchangeApi& mea)
    : fail_callback_(std::move(fail_callback)),
      running_(false),
      reset_connection_(false),
      subscribe_done_(false),
      spec_(spec),
      policy_(std::move(policy)),
      ioc_(ioc),
      strand_(*ioc_),
      ctx_(boost::asio::ssl::context::sslv23_client),
      resolver_(strand_),
      use_ssl_(spec.use_ssl),
      mea_(mea) {
  if (use_ssl_) {
    ws_ssl_.reset(new WebsocketSsl(strand_, ctx_));
  } else {
    ws_nossl_.reset(new WebsocketNoSsl(strand_));
  }
}

BasePrivateWsConnection::~BasePrivateWsConnection() {}

void BasePrivateWsConnection::RunAsync() {
  if (mea_.exchange == coin2::exchange::base::market::ExchangeType::Bithumb) {
    return;
  }
  if (!IsValidSpec()) {
    LOG(WARNING) << "private ws subscriber disabled with invalid spec !!";
    return;
  }
  running_ = true;
  boost::asio::spawn(
      strand_,
      std::bind(&BasePrivateWsConnection::Run, shared_from_this(), std::placeholders::_1));
}

std::string BasePrivateWsConnection::Decompress(const std::string& s) const {
  if (spec_.decompress == "deflate") {
    return deflate_string(s);
  } else if (spec_.decompress == "gzip") {
    return decompress_gzip_string(s);
  } else {
    return s;
  }
}

void BasePrivateWsConnection::Fail(
    boost::asio::yield_context* yield,
    boost::system::error_code ec,
    const std::string& what) {
  LOG(ERROR) << "[BasePrivateWsConnection Error] " << what << ": " << ec.message();

  // publish connection reset message to order processor
  BaseParser::RawMessage raw_msg;
  raw_msg.type = coin::proto::OrderSubTopicId::SYSTEM;
  raw_msg.timestamp = GetCurrentTimestamp();

  nlohmann::json json;
  json["value"] = ec.value();
  json["what"] = what;
  json["message"] = ec.message();
  json["category"] = ec.category().name();
  json["mea"] = mea_.String();
  json["error_code"] = "WS_CONNECTION_RESET";

  raw_msg.error_code.set_error_code(0);
  raw_msg.error_code.set_error_message(json.dump());
  raw_msg.error_code.set_category("WS_PRIVATE_CLIENT");
  processor()->ProcessRawMessage(raw_msg);

  if (!running_) {
    return;
  }
  running_ = false;

  // until async_close is done, the coroutine is alive
  if (use_ssl_) {
    ws_ssl_->async_close(boost::beast::websocket::close_code::normal, (*yield)[ec]);
  } else {
    ws_nossl_->async_close(boost::beast::websocket::close_code::normal, (*yield)[ec]);
  }

  if (ec) {
    LOG(ERROR) << "[BasePrivateWsConnection Error] async_close() failed";
  }

  // sleep
  boost::asio::steady_timer timer(*ioc_);
  timer.expires_from_now(std::chrono::seconds(5));
  timer.async_wait((*yield)[ec]);
  if (ec) {
    LOG(ERROR) << "[BasePrivateWsConnection Error]  sleep failed";
  }

  // now we have no websocket operation scheduled.
  // teardown
  boost::asio::post(*ioc_, std::bind(fail_callback_));
}

void BasePrivateWsConnection::Subscribe(boost::asio::yield_context yield) {
  using coin2::exchange::base::market::ExchangeType;
  boost::system::error_code ec;
  const auto requests_ = policy_->GetSubscriptionMsgs();

  boost::asio::steady_timer timer(*ioc_);
  for (const auto& req_origin : requests_) {
    auto req = req_origin;
    if (mea_.exchange == ExchangeType::Kraken || mea_.exchange == ExchangeType::Lbank ||
        mea_.exchange == ExchangeType::Bit) {
      boost::replace_all(req, "{}", listen_key_);
    } else if (mea_.exchange == ExchangeType::Gateio) {
      if (mea_.market == MarketType::Futures) {
        boost::replace_all(req, "%2", user_id_);
      }
    }
    LOG(INFO) << "Sending Sub Request " << req;
    if (use_ssl_) {
      ws_ssl_->async_write(boost::asio::buffer(req), yield[ec]);
    } else {
      ws_nossl_->async_write(boost::asio::buffer(req), yield[ec]);
    }
    if (ec) {
      return Fail(&yield, ec, "sending subscription request fail");
    }
    timer.expires_from_now(std::chrono::seconds(1));
    timer.async_wait(yield[ec]);
  }
  subscribe_done_ = true;

  // if we must ping manually
  if (!spec_.handle_server_ping) {
    while (running_) {
      std::string ping = policy_->GetPingMsg();
      LOG(INFO) << "PING";
      if (use_ssl_) {
        ws_ssl_->async_write(boost::asio::buffer(ping), yield[ec]);
      } else {
        ws_nossl_->async_write(boost::asio::buffer(ping), yield[ec]);
      }
      if (ec) {
        return Fail(&yield, ec, "sending ping fail");
      }
      timer.expires_from_now(std::chrono::seconds(spec_.ping_period_sec));
      timer.async_wait(yield[ec]);
    }
  }
}

void BasePrivateWsConnection::Run(boost::asio::yield_context yield) {
  using coin2::exchange::base::market::ExchangeType;
  auto spec_path = spec_.path;
  if (mea_.exchange == ExchangeType::Binance || mea_.exchange == ExchangeType::Apollox ||
      mea_.exchange == ExchangeType::Kraken || mea_.exchange == ExchangeType::Lbank ||
      mea_.exchange == ExchangeType::Bit || mea_.exchange == ExchangeType::Mexc ||
      mea_.exchange == ExchangeType::Bittrex) {
    const auto check_period = boost::posix_time::milliseconds(1000);
    // Wait until listen_key is ready
    boost::asio::deadline_timer timer(*ioc_);
    while (!HasListenKey()) {
      LOG(INFO) << "ListenKey not ready!";
      timer.expires_from_now(check_period);
      timer.async_wait(yield);
    }
    LOG(INFO) << "listen_key ready: " << listen_key_;
    if (mea_.exchange == ExchangeType::Binance || mea_.exchange == ExchangeType::Apollox) {
      spec_path += "/" + listen_key_;
    }
    if (mea_.exchange == ExchangeType::Mexc || mea_.exchange == ExchangeType::Bittrex) {
      boost::replace_all(spec_path, "{}", listen_key_);
    }
  } else if (mea_.exchange == ExchangeType::Gopax) {
    auto auth = GopaxWsAuth(policy_->GetOrderConnectionConfig().key_filepath());
    spec_path = auth.AppendAuthParams(spec_path);
  } else if (mea_.exchange == ExchangeType::Gateio) {
    if (mea_.market == MarketType::Futures) {
      const auto check_period = boost::posix_time::milliseconds(1000);
      // Wait until user_id is ready
      boost::asio::deadline_timer timer(*ioc_);
      while (!HasUserId()) {
        LOG(INFO) << "user id not ready!";
        timer.expires_from_now(check_period);
        timer.async_wait(yield);
      }
      LOG(INFO) << "user id ready: " << user_id_;
    }
  } else if (mea_.exchange == ExchangeType::Kucoin) {
    const auto check_period = boost::posix_time::milliseconds(1000);
    // Wait until user_id is ready
    boost::asio::deadline_timer timer(*ioc_);
    while (!HasBullet()) {
      LOG(INFO) << "bullet not ready!";
      timer.expires_from_now(check_period);
      timer.async_wait(yield);
    }
    LOG(INFO) << "bullet is ready: size = " << bullet_info_.size();
    std::string endpoint = bullet_info_[0];
    std::string host;
    std::string path;
    auto pos_left = endpoint.find("//");
    auto pos_right = endpoint.rfind('/');
    if ((pos_left != std::string::npos) && (pos_right != std::string::npos)) {
      host = endpoint.substr(pos_left + 2, pos_right - pos_left - 2);
      path = endpoint.substr(pos_right);
      path += "?token=" + bullet_info_[4];
    }
    spec_.host = host;
    spec_.path = spec_path = path;
    spec_.use_ssl = use_ssl_ = (bullet_info_[1] == "true") ? true : false;
    spec_.ping_period_sec = (int)(stoll(bullet_info_[2]) / 1000 / 2);
    spec_.timeout_sec = (int)(stoll(bullet_info_[3]) / 1000);
  }

  boost::system::error_code ec;
  {
    // Look up the domain name
    auto const ip = resolver_.async_resolve(spec_.host, spec_.port, yield[ec]);
    if (ec) {
      return Fail(
          &yield,
          ec,
          std::string("cannot resolve host=") + spec_.host + " port=" + spec_.port);
    }

    // Make the connection on the IP address we get from a lookup
    if (use_ssl_) {
      boost::asio::async_connect(
          ws_ssl_->next_layer().next_layer(),
          ip.begin(),
          ip.end(),
          yield[ec]);
    } else {
      // TODO(jshin): does this work..?
      boost::asio::async_connect(
          ws_nossl_->next_layer().lowest_layer(),
          ip.begin(),
          ip.end(),
          yield[ec]);
    }
    if (ec) {
      return Fail(&yield, ec, std::string("cannot connect ip"));
    }
  }

  std::string prex_api_key;
  std::string prex_api_expires;
  std::string prex_api_sign;
  std::string prex_auth_msg;
  if (mea_.exchange == ExchangeType::Prex) {
    auto prex_auth = PrexFuturesWsApiKeyAuth(policy_->GetOrderConnectionConfig().key_filepath());
    prex_auth.CreateSignature(&prex_api_key, &prex_api_expires, &prex_api_sign, &prex_auth_msg);
  }
  auto decorator = [&](auto& m) {
    if (mea_.exchange == ExchangeType::Prex) {
      m.insert("api-key", prex_api_key);
      m.insert("api-expires", prex_api_expires);
      m.insert("api-signature", prex_api_sign);
    }
  };

  if (use_ssl_) {
    if (!SSL_set_tlsext_host_name(ws_ssl_->next_layer().native_handle(), spec_.host.data())) {
      boost::system::error_code ec{
          static_cast<int>(::ERR_get_error()),
          boost::asio::error::get_ssl_category()};
      return Fail(&yield, ec, ec.message());
    }
    LOG(INFO) << "Set SNI Hostname. host = " << spec_.host;

    LOG(INFO) << "Start ssl hand shake!";
    ws_ssl_->set_option(boost::beast::websocket::stream_base::decorator(decorator));
    ws_ssl_->next_layer().async_handshake(boost::asio::ssl::stream_base::client, yield[ec]);
    if (ec) {
      return Fail(&yield, ec, "ssl_handshake");
    }
    ws_ssl_->async_handshake(spec_.host, spec_path, yield[ec]);
  } else {
    ws_nossl_->set_option(boost::beast::websocket::stream_base::decorator(decorator));
    ws_nossl_->async_handshake(spec_.host, spec_path, yield[ec]);
  }
  if (ec) {
    LOG(INFO) << "Websocket hand shake failed!";
    return Fail(&yield, ec, "handshake");
  }

  bool support_ping_frame = false;
  if (mea_.exchange == ExchangeType::Dydx || mea_.exchange == ExchangeType::Crypto ||
      mea_.exchange == ExchangeType::Bitmart || mea_.exchange == ExchangeType::Bit ||
      mea_.exchange == ExchangeType::Maicoin || mea_.exchange == ExchangeType::Bittrex) {
    support_ping_frame = true;
  }

  if (use_ssl_) {
    ws_ssl_->set_option(boost::beast::websocket::stream_base::timeout{
        std::chrono::seconds(5),                  // handshake timeout
        std::chrono::seconds(spec_.timeout_sec),  // idle timeout
        support_ping_frame});
  } else {
    ws_nossl_->set_option(boost::beast::websocket::stream_base::timeout{
        std::chrono::seconds(5),                  // handshake timeout
        std::chrono::seconds(spec_.timeout_sec),  // idle timeout
        support_ping_frame});
  }

  LOG(INFO) << "LOGIN";
  std::size_t bytes_transferred;

  if ((mea_.exchange == ExchangeType::Gopax) || (mea_.exchange == ExchangeType::Gateio) ||
      (mea_.exchange == ExchangeType::Gdax) || (mea_.exchange == ExchangeType::Mexc) ||
      (mea_.exchange == ExchangeType::Kucoin) || (mea_.exchange == ExchangeType::Kraken) ||
      (mea_.exchange == ExchangeType::Lbank) || (mea_.exchange == ExchangeType::Bit)) {
    boost::asio::spawn(
        strand_,
        std::bind(&BasePrivateWsConnection::Subscribe, shared_from_this(), std::placeholders::_1));
  } else if (mea_.exchange != ExchangeType::Binance && mea_.exchange != ExchangeType::Apollox) {
    std::string req;
    if (mea_.exchange == ExchangeType::Prex) {
      req = prex_auth_msg;
    } else {
      req = policy_->GetAuthMsg();
    }
    int repeat_time = 1;
    if (mea_.exchange == ExchangeType::Ftx) {
      repeat_time = 2;
    }
    for (int i = 0; i < repeat_time; i++) {
      if (use_ssl_) {
        ws_ssl_->async_write(boost::asio::buffer(req), yield[ec]);
      } else {
        ws_nossl_->async_write(boost::asio::buffer(req), yield[ec]);
      }
      if (ec) {
        return Fail(&yield, ec, "sending login request fail");
      }
    }

    std::string s;
    if (mea_.exchange == ExchangeType::Bittrex) {
      do {
        if (!s.empty()) {
          LOG(INFO) << "ignore websocket msg:" << s;
        }
        if (use_ssl_) {
          bytes_transferred = ws_ssl_->async_read(buffer_, yield[ec]);
        } else {
          bytes_transferred = ws_nossl_->async_read(buffer_, yield[ec]);
        }
        if (ec) {
          return Fail(&yield, ec, "fail reading ws auth");
        }
        boost::ignore_unused(bytes_transferred);
        s = boost::beast::buffers_to_string(buffer_.data());
        buffer_.consume(buffer_.size());
      } while (s.find("\"I\":\"0\"") == std::string::npos);
    } else {
      if (use_ssl_) {
        bytes_transferred = ws_ssl_->async_read(buffer_, yield[ec]);
      } else {
        bytes_transferred = ws_nossl_->async_read(buffer_, yield[ec]);
      }
      if (ec) {
        return Fail(&yield, ec, "fail reading ws auth");
      }
      boost::ignore_unused(bytes_transferred);
      s = boost::beast::buffers_to_string(buffer_.data());
      buffer_.consume(buffer_.size());
    }
    if (mea_.exchange != ExchangeType::Bitmart) {
      s = Decompress(s);
    }
    LOG(INFO) << "[Ws Login resp] " << s;
    if (!policy_->IsSuccessAuthResp(s)) {
      return Fail(
          &yield,
          boost::system::errc::make_error_code(boost::system::errc::operation_not_permitted),
          "Ws Auth Fail");
    }

    boost::asio::spawn(
        strand_,
        std::bind(&BasePrivateWsConnection::Subscribe, shared_from_this(), std::placeholders::_1));
  }

  // publish connection reset message to order processor
  BaseParser::RawMessage raw_msg;
  raw_msg.type = coin::proto::OrderSubTopicId::SYSTEM;
  raw_msg.timestamp = GetCurrentTimestamp();

  // notify OG
  nlohmann::json json;
  json["mea"] = mea_.String();
  json["error_code"] = "WS_CONNECTED";
  raw_msg.error_code.set_error_code(0);
  raw_msg.error_code.set_error_message(json.dump());
  raw_msg.error_code.set_category("WS_PRIVATE_CLIENT");
  processor()->ProcessRawMessage(raw_msg);

  boost::asio::spawn(
      strand_,
      std::bind(&BasePrivateWsConnection::Monitor, shared_from_this(), std::placeholders::_1));

  while (running_) {
    if (use_ssl_) {
      bytes_transferred = ws_ssl_->async_read(buffer_, yield[ec]);
    } else {
      bytes_transferred = ws_nossl_->async_read(buffer_, yield[ec]);
    }
    if (ec) {
      return Fail(&yield, ec, "fail reading ws msg");
    }
    boost::ignore_unused(bytes_transferred);
    std::string s;
    if (mea_.exchange == ExchangeType::Bitmart) {
      // for debug
      try {
        s = Decompress(boost::beast::buffers_to_string(buffer_.data()));
      } catch (const std::exception& e) {
        LOG(ERROR) << "[" << __func__
                   << "] original message: " << boost::beast::buffers_to_string(buffer_.data());
        LOG(ERROR) << "[" << __func__ << "] exception: " << e.what() << ". force reconnect";
        ec = boost::system::errc::make_error_code(boost::system::errc::bad_message);
        return Fail(&yield, ec, "force reconnect...");
      } catch (...) {
        LOG(ERROR) << "[" << __func__
                   << "] original message: " << boost::beast::buffers_to_string(buffer_.data());
        LOG(ERROR) << "[" << __func__ << "] exception. force reconnect";
        ec = boost::system::errc::make_error_code(boost::system::errc::bad_message);
        return Fail(&yield, ec, "force reconnect...");
      }
    } else {
      s = Decompress(boost::beast::buffers_to_string(buffer_.data()));
    }
    buffer_.consume(buffer_.size());

    if (s == "pong" || s == "PONG") {
      // ignore
      continue;
    }

    if (CheckIfNeedReconnect(s)) {
      return Fail(&yield, ec, "force reconnect...");
    }

    // if server send ping first (eg. huobi),
    // linchuan: if ping and subscribe at same time will crash
    if (const auto pong = policy_->MakePingMsgFromServer(s)) {
      if (spec_.handle_server_ping && subscribe_done_) {
        DLOG(INFO) << "PING";
        if (use_ssl_) {
          ws_ssl_->async_write(boost::asio::buffer(pong.value()), yield[ec]);
        } else {
          ws_nossl_->async_write(boost::asio::buffer(pong.value()), yield[ec]);
        }
        if (ec) {
          return Fail(&yield, ec, "pong");
        }
      }
      continue;
    }

    if (policy_->IsConnectionErrorMsg(s)) {
      auto ec = make_error_code(boost::system::errc::timed_out);
      LOG(ERROR) << "ws_error_msg received:" << s;
      return Fail(&yield, ec, "connection error msg");
    }

    if (mea_.exchange == ExchangeType::Bittrex &&
        s.find("authenticationExpiring") != std::string::npos) {
      LOG(INFO) << "auth expiring, relogin.";

      std::string req = policy_->GetAuthMsg();

      if (use_ssl_) {
        ws_ssl_->async_write(boost::asio::buffer(req), yield[ec]);
      } else {
        ws_nossl_->async_write(boost::asio::buffer(req), yield[ec]);
      }
      if (ec) {
        return Fail(&yield, ec, "sending login request fail");
      }
      continue;
    }

    // handle income message
    BaseParser::RawMessage msg;
    msg.type = OrderSubTopicId::WS_MESSAGE;
    msg.data = &s;
    msg.timestamp = GetCurrentTimestamp();
    processor()->ProcessRawMessage(msg);
  }

  // Close the WebSocket connection
  if (use_ssl_) {
    ws_ssl_->async_close(boost::beast::websocket::close_code::normal, yield[ec]);
  } else {
    ws_nossl_->async_close(boost::beast::websocket::close_code::normal, yield[ec]);
  }
  if (ec) {
    return Fail(&yield, ec, "close");
  }
}

bool BasePrivateWsConnection::CheckIfNeedReconnect(std::string_view msg_view) {
  if (mea_.exchange == ExchangeType::Binance &&
      std::string_view(msg_view.data(), 50).find("listenKeyExpired") != std::string::npos) {
    SPDLOG_ERROR("[Binance] listen key is expired\n{}", msg_view);
    return true;
  }
  return false;
}

auto BasePrivateWsConnection::MakeMockPingMsg() -> std::optional<std::string> {
  auto mock_ping = policy_->GetMockPingMsg();
  if (!mock_ping) {
    return mock_ping;
  }
  if (ExchangeType::Binance == mea_.exchange && MarketType::Futures == mea_.market) {
    boost::replace_all(*mock_ping, "{}", listen_key_);
  }
  return mock_ping;
}

void BasePrivateWsConnection::Monitor(boost::asio::yield_context yield) {
  boost::asio::steady_timer timer(*ioc_);
  boost::system::error_code ec;
  last_recv_msg_ts_ = GetCurrentTimestamp();

  auto send_mock_ping = [&, this]() {
    auto mock_ping = MakeMockPingMsg();
    if (!mock_ping) {
      return;
    }
    SPDLOG_INFO("sending mock ping: {}", *mock_ping);
    if (use_ssl_) {
      ws_ssl_->async_write(boost::asio::buffer(*mock_ping), yield[ec]);
    } else {
      ws_nossl_->async_write(boost::asio::buffer(*mock_ping), yield[ec]);
    }
    if (ec) {
      Fail(&yield, ec, "failed to send mock ping message");
    }
  };

  while (running_) {
    if (reset_connection_) {
      ec = boost::system::errc::make_error_code(boost::system::errc::operation_canceled);
      Fail(&yield, ec, "reset connection by demand");
      break;
    }
    if (mea_.exchange == ExchangeType::Binance) {  // TODO(daniel): apply to all exchanges
      auto now = GetCurrentTimestamp();
      if (last_recv_msg_ts_ + 300'000'000'000LL /* 5 mins */ < now) {
        ec = boost::system::errc::make_error_code(boost::system::errc::operation_canceled);
        Fail(&yield, ec, "reset connection due to there is no message push in 5 mins");
        break;
      }
      if (last_recv_msg_ts_ + 120'000'000'000LL /* 2 mins */ < now) {
        send_mock_ping();
      }
    }

    timer.expires_from_now(std::chrono::seconds(5));
    timer.async_wait(yield[ec]);
  }
}

namespace {

using ExchangeType = base::market::ExchangeType;
using MarketType = coin2::exchange::base::market::MarketType;

std::string ExtractLowerUnderlying(const std::string& str) {
  std::string data = str.substr(0, 3);
  std::transform(data.begin(), data.end(), data.begin(), [](unsigned char c) {
    return std::tolower(c);
  });
  return data;
}

}  // namespace

BasePrivateWsSubscriberPolicy::BasePrivateWsSubscriberPolicy(
    const MarketExchangeApi& mea,
    const OrderConnectionConfig& conn_config,
    const std::set<std::string>& native_symbols)
    : mea_(mea), conn_config_(conn_config), native_symbols_(native_symbols) {
  std::ifstream ifs("data/coin2/order/og_ws_info.json");
  nlohmann::json jf = nlohmann::json::parse(ifs);
  if (jf.find(mea.String()) != jf.end()) {
    ws_config_json_ = jf[mea.String()];
  }
}

std::vector<std::string> BasePrivateWsSubscriberPolicy::GetSubscriptionMsgs() const {
  std::vector<std::string> reqs;
  if (mea_.exchange == ExchangeType::Bitflyer) {
    std::string ts_str = std::to_string(GetCurrentTimestamp() / 1000000);
    reqs.push_back(fmt::format(
        R"({{"jsonrpc": "2.0", "id": "{}", "method": "subscribe", )"
        R"("params": {{"channel": "child_order_events"}}}})",
        ts_str));
    return reqs;
  }
  if (mea_.exchange == ExchangeType::Gdax) {
    GdaxWsAuth auth = GdaxWsAuth(conn_config_.key_filepath());
    auto sign_list = auth.GetSignatureStringList();
    std::vector<std::string> sign_vector{
        std::make_move_iterator(std::begin(sign_list)),
        std::make_move_iterator(std::end(sign_list))};
    for (const auto& native_symbol : native_symbols_) {
      reqs.push_back(fmt::format(
          R"({{"type": "subscribe", "product_ids": ["{}"], "channels": ["user"], )"
          R"("key": "{}", "passphrase": "{}", "timestamp": "{}", "signature": "{}"}})",
          native_symbol,
          sign_vector[0],
          sign_vector[1],
          sign_vector[2],
          sign_vector[3]));
    }
    if (!native_symbols_.empty()) {
      // need to send auth subscribe first
      reqs.push_back(
          R"({"type":"subscribe","channels":[{"name":"heartbeat","product_ids":["BTC-USD"]}]})");
    }
    return reqs;
  }
  if (mea_.exchange == ExchangeType::Huobi && mea_.market == MarketType::Futures) {
    std::string topic_str;
    std::string cross_mode_str = "";
    if (conn_config_.use_cross_mode()) {
      cross_mode_str = "_cross";
      reqs.push_back(
          R"({"op": "sub", "topic": "accounts_cross.USDT", "cid": "accounts_cross.USDT"})");
    }
    for (const auto& native_symbol : native_symbols_) {
      if (mea_ == MarketExchangeApi{MarketType::Futures, ExchangeType::Huobi, "v1-swap"} ||
          mea_ == MarketExchangeApi{MarketType::Futures, ExchangeType::Huobi, "v1-linear-swap"}) {
        topic_str = native_symbol;
      } else {
        topic_str = ExtractLowerUnderlying(native_symbol);
      }
      if (!conn_config_.use_cross_mode()) {
        reqs.push_back(fmt::format(
            R"({{"op": "sub", "topic": "accounts.{}", "cid": "accounts.{}"}})",
            topic_str,
            native_symbol));
      }
      reqs.push_back(fmt::format(
          R"({{"op": "sub", "topic": "positions{}.{}", "cid": "positions.{}"}})",
          cross_mode_str,
          topic_str,
          native_symbol));
      reqs.push_back(fmt::format(
          R"({{"op": "sub", "topic": "orders{}.{}", "cid": "orders.{}"}})",
          cross_mode_str,
          topic_str,
          native_symbol));
    }
    return reqs;
  }
  if (mea_.exchange == ExchangeType::Kraken || mea_.exchange == ExchangeType::Lbank ||
      mea_.exchange == ExchangeType::Gopax || mea_.exchange == ExchangeType::Bit ||
      mea_.exchange == ExchangeType::Mexc) {
    for (auto& msg : ws_config_json_["subscription_msgs"]) {
      reqs.push_back(msg.get<std::string>());
    }
    return reqs;
  }
  if (mea_.exchange == ExchangeType::Deribit && mea_.market == coin::proto::Futures) {
    static const std::set<std::string> deribit_futures_currencies =
        {"BTC", "ETH", "USDC", "ETHW", "SOL"};
    std::map<std::string, std::vector<std::string>> channels;
    nlohmann::json json;
    json["method"] = "private/subscribe";
    json["jsonrpc"] = "2.0";
    json["id"] = 100;

    std::set<std::string> native_currencies;
    for (const auto& native_symbol : native_symbols_) {
      // TODO(daniel): too specify here
      auto vec = Split(native_symbol, "-");
      std::string native_currency = vec[0];
      auto pos = native_currency.find("_");
      if (pos != std::string::npos) {
        auto vec = Split(native_currency, "_");
        native_currency = vec[1];
      }
      if (deribit_futures_currencies.count(native_currency) == 0) {
        continue;
      }
      native_currencies.emplace(native_currency);
    }
    channels["channels"] = {};
    for (const auto& native_currency : native_currencies) {
      for (auto& msg : ws_config_json_["subscription_msgs"]) {
        auto msg_str = msg.get<std::string>();
        channels["channels"].push_back(fmt::format(msg, native_currency));
      }
    }
    json["params"] = channels;
    reqs.push_back(json.dump());
    return reqs;
  }
  if (mea_.exchange == ExchangeType::Deribit && mea_.market == coin::proto::Options) {
    static const std::set<std::string> deribit_options_currencies =
        {"BTC", "ETH", "USDC", "ETHW", "SOL"};
    std::map<std::string, std::vector<std::string>> channels;
    nlohmann::json json;
    json["method"] = "private/subscribe";
    json["jsonrpc"] = "2.0";
    json["id"] = 100;

    std::set<std::string> native_currencies;
    for (const auto& native_symbol : native_symbols_) {
      auto vec = Split(native_symbol, "-");
      std::string native_currency = vec[0];
      if (deribit_options_currencies.count(native_currency) == 0) {
        continue;
      }
      native_currencies.emplace(native_currency);
    }
    channels["channels"] = {};
    for (const auto& native_currency : native_currencies) {
      for (auto& msg : ws_config_json_["subscription_msgs"]) {
        auto msg_str = msg.get<std::string>();
        channels["channels"].push_back(fmt::format(msg, native_currency));
      }
    }
    json["params"] = channels;
    reqs.push_back(json.dump());
    LOG(INFO) << "Deribit subscription: " << json.dump();
    return reqs;
  }
  if (mea_.exchange == ExchangeType::Gateio) {
    if (mea_.market == MarketType::Spot) {
      GateioWsAuth auth = GateioWsAuth(conn_config_.key_filepath());
      for (auto& msg : ws_config_json_["subscription_msgs"]) {
        std::string msg_str = msg.get<std::string>();
        auto j = nlohmann::json::parse(msg_str);
        std::string channel = j["channel"].get<std::string>();
        std::string event = j["event"].get<std::string>();
        int64_t time = GetCurrentTimestamp() / 1e9;
        std::string time_str = std::to_string(time);
        auto sign = auth.CreateSignature(channel, event, time);
        std::string symbols;
        for (const auto& native_symbol : native_symbols_) {
          symbols += "\"" + native_symbol + "\",";
        }
        symbols = symbols.substr(0, symbols.size() - 1);
        boost::replace_all(msg_str, "\"%1\"", time_str);
        boost::replace_all(msg_str, "\"%2\"", symbols);
        boost::replace_all(msg_str, "\"%3\"", sign);
        reqs.push_back(msg_str);
      }
    } else if (mea_.market == MarketType::Futures) {
      GateioWsAuth auth = GateioWsAuth(conn_config_.key_filepath());
      for (auto& msg : ws_config_json_["subscription_msgs"]) {
        std::string msg_str = msg.get<std::string>();
        auto j = nlohmann::json::parse(msg_str);
        std::string channel = j["channel"].get<std::string>();
        std::string event = j["event"].get<std::string>();
        int64_t time = GetCurrentTimestamp() / 1e9;
        std::string time_str = std::to_string(time);
        auto sign = auth.CreateSignature(channel, event, time);
        for (const auto& native_symbol : native_symbols_) {
          boost::replace_all(msg_str, "\"%1\"", time_str);
          boost::replace_all(msg_str, "%3", native_symbol);
          boost::replace_all(msg_str, "\"%4\"", sign);
          reqs.push_back(msg_str);
        }
      }
    }
    return reqs;
  }
  if (mea_.exchange == ExchangeType::Kucoin) {
    if (mea_.market == MarketType::Futures) {
      for (auto& msg : ws_config_json_["subscription_msgs"]) {
        std::string msg_str = msg.get<std::string>();
        std::string symbols;
        for (const auto& native_symbol : native_symbols_) {
          symbols += native_symbol + ",";
        }
        symbols = symbols.substr(0, symbols.size() - 1);
        boost::replace_all(msg_str, "{}", symbols);
        reqs.push_back(msg_str);
      }
      return reqs;
    }
  }
  if (mea_.exchange == ExchangeType::Crypto) {
    auto auth = CryptoWsAuth(conn_config_.key_filepath());
    {
      // balance
      std::set<std::string> balance_channels = {"user.balance"};
      reqs.push_back(auth.GenerateRequest(balance_channels));
    }
    {
      // TODO(daniel): if too many products traded
      // order
      // trade
      std::set<std::string> order_channels;
      std::set<std::string> trade_channels;
      for (const auto& native_symbol : native_symbols_) {
        order_channels.emplace(fmt::format("user.order.{}", native_symbol));
        trade_channels.emplace(fmt::format("user.trade.{}", native_symbol));
      }
      reqs.push_back(auth.GenerateRequest(order_channels));
      reqs.push_back(auth.GenerateRequest(trade_channels));
    }
    return reqs;
  }

  for (auto& msg : ws_config_json_["subscription_msgs"]) {
    auto msg_str = msg.get<std::string>();
    if (msg_str.find("{}") != std::string::npos) {
      for (const auto& native_symbol : native_symbols_) {
        auto tmp_str = msg_str;
        boost::replace_all(tmp_str, "{}", native_symbol);
        reqs.push_back(tmp_str);
      }
    } else {
      reqs.push_back(msg_str);
    }
  }
  return reqs;
}

std::string BasePrivateWsSubscriberPolicy::GetAuthMsg() const {
  if (mea_.exchange == ExchangeType::Bitflyer) {
    BitflyerWsAuth auth = BitflyerWsAuth(conn_config_.key_filepath());
    return auth.CreateSignature();
  }
  if (mea_.exchange == ExchangeType::Bitmex) {
    BitmexWsAuth auth = BitmexWsAuth(conn_config_.key_filepath());
    return auth.CreateSignature();
  }
  if (mea_.exchange == ExchangeType::Bybit) {
    if (mea_.market == MarketType::Futures) {
      BybitWsAuth auth = BybitWsAuth(conn_config_.key_filepath());
      return auth.CreateSignature();
    } else if (mea_.market == MarketType::Spot) {
      BybitSpotWsAuth auth = BybitSpotWsAuth(conn_config_.key_filepath());
      return auth.CreateSignature();
    }
  }
  if (mea_.exchange == ExchangeType::Bitmart) {
    BitmartWsAuth auth(conn_config_.key_filepath());
    return auth.CreateSignature();
  }
  if (mea_.exchange == ExchangeType::Ftx) {
    FtxWsAuth auth = FtxWsAuth(conn_config_.key_filepath());
    return auth.CreateSignature();
  }
  if (mea_.exchange == ExchangeType::Huobi) {
    HuobiWsAuth auth = HuobiWsAuth(conn_config_.key_filepath());
    if (mea_.market == MarketType::Spot) {
      return auth.CreateSignatureWs2(conn_config_.ws_host(), ws_config_json_["path"]);
    } else {
      return auth.CreateSignature(conn_config_.ws_host(), ws_config_json_["path"]);
    }
  }
  if (mea_.exchange == ExchangeType::Okex) {
    OkexWsAuth auth = OkexWsAuth(conn_config_.key_filepath());
    return auth.CreateSignature(mea_.api);
  }
  if (mea_.exchange == ExchangeType::Prex) {
    // auto auth = PrexFuturesWsApiKeyAuth(conn_config_.key_filepath());
    // return auth.CreateSignature();
    return "";
  }
  if (mea_.exchange == ExchangeType::Deribit) {
    auto auth = AuthKey::FromFile(conn_config_.key_filepath());
    std::map<std::string, std::string> params;
    params["grant_type"] = "client_credentials";
    params["scope"] = "session:cpp_og";
    params["client_id"] = auth.access_key();
    params["client_secret"] = auth.secret_key();

    nlohmann::json json;
    json["id"] = 0;
    json["method"] = "public/auth";
    json["jsonrpc"] = "2.0";
    json["params"] = params;
    return json.dump();
  }
  if (mea_.exchange == ExchangeType::Phemex) {
    PhemexWsAuth auth = PhemexWsAuth(conn_config_.key_filepath());
    return auth.CreateSignature();
  }
  if (mea_.exchange == ExchangeType::Dydx) {
    auto auth = DydxWsAuth(conn_config_.key_filepath());
    return auth.CreateSignature();
  }
  if (mea_.exchange == ExchangeType::Crypto) {
    auto auth = CryptoWsAuth(conn_config_.key_filepath());
    return auth.CreateSignature();
  }
  if (mea_.exchange == ExchangeType::Bitget) {
    auto auth = BitgetWsAuth(conn_config_.key_filepath());
    return auth.CreateSignature();
  }
  if (mea_.exchange == ExchangeType::Maicoin) {
    auto auth = MaicoinWsAuth(conn_config_.key_filepath());
    return auth.CreateSignature();
  }
  if (mea_.exchange == ExchangeType::Bittrex) {
    auto auth = BittrexWsAuth(conn_config_.key_filepath());
    return auth.CreateSignature();
  }

  return "";
}

std::string BasePrivateWsSubscriberPolicy::GetPingMsg() const {
  if (mea_.exchange == ExchangeType::Gateio) {
    if (mea_.market == MarketType::Spot) {
      std::string ping_msg = ws_config_json_["ping_msg"];
      int64_t time = GetCurrentTimestamp() / 1e9;
      std::string time_str = std::to_string(time);
      boost::replace_all(ping_msg, "\"%1\"", time_str);
      return ping_msg;
    } else if (mea_.market == MarketType::Futures) {
      std::string ping_msg = ws_config_json_["ping_msg"];
      int64_t time = GetCurrentTimestamp() / 1e9;
      std::string time_str = std::to_string(time);
      boost::replace_all(ping_msg, "\"%1\"", time_str);
      return ping_msg;
    }
  } else if (mea_.exchange == ExchangeType::Bybit || mea_.exchange == ExchangeType::Lbank) {
    if (mea_.market == MarketType::Spot) {
      std::string ping_msg = ws_config_json_["ping_msg"];
      int64_t time = GetCurrentTimestamp() / 1000000;
      std::string time_str = std::to_string(time);
      boost::replace_all(ping_msg, "{}", time_str);
      return ping_msg;
    }
  }

  return ws_config_json_["ping_msg"];
}

bool BasePrivateWsSubscriberPolicy::IsConnectionErrorMsg(const std::string& resp) const {
  if (ws_config_json_.contains("connection_error")) {
    const std::string err = ws_config_json_["connection_error"].get<std::string>();
    return resp.find(err) != std::string::npos;
  } else {
    // always considered not an error.
    return false;
  }
}

std::optional<std::string> BasePrivateWsSubscriberPolicy::MakePingMsgFromServer(
    const std::string& server_ping) const {
  if (mea_.exchange == ExchangeType::Huobi) {
    std::string find_str;
    if (mea_.market == MarketType::Spot) {
      find_str = R"("action":"ping")";
    } else {
      find_str = R"("op":"ping")";
    }
    if (server_ping.find(find_str) != std::string::npos) {
      return get_pong(server_ping);
    }
  } else if (mea_.exchange == ExchangeType::Gopax) {
    std::string find_str = R"(primus::ping::)";
    if (server_ping.find(find_str) != std::string::npos) {
      return get_pong(server_ping);
    }
  } else if (mea_.exchange == ExchangeType::Crypto) {
    if (server_ping.find(R"(public/heartbeat)") != std::string::npos) {
      return CryptoWsAuth::GeneratePong(server_ping);
    }
  }
  return {};
}

bool BasePrivateWsSubscriberPolicy::IsSuccessAuthResp(const std::string& resp) const {
  return resp.find(ws_config_json_["success_auth"]) != std::string::npos;
}

auto BasePrivateWsSubscriberPolicy::GetMockPingMsg() const -> std::optional<std::string> {
  if (ws_config_json_.find("mock_ping") != ws_config_json_.end()) {
    return ws_config_json_["mock_ping"].get<std::string>();
  }
  return {};
}

}  // namespace coin2::exchange::base::order
